package com.it.sink;

import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 自定义sink function
 *
 * @author code1997
 */
public class ConsumerHBaseSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment);
        //eventSource.assignTimestampsAndWatermarks();
        String sql = "insert into flink_demo.clicks values(?,?)";
        eventSource.addSink(JdbcSink.sink(sql, new JdbcStatementBuilder<Event>() {
            @Override
            public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
                preparedStatement.setString(1, event.user);
                preparedStatement.setString(2, event.url);
            }
        }, new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:mysql://docker_server:3306/flink_demo?useSSL=true&useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai")
                .withDriverName("com.mysql.cj.jdbc.Driver")
                .withUsername("root")
                .withPassword("19971001")
                .build()));
        executionEnvironment.execute();

    }
}
